package com.cskaoyan.mall.promo.mq.producer;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import jdk.nashorn.internal.ir.ReturnNode;
import org.apache.commons.io.filefilter.RegexFileFilter;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @author: 杨庆
 * @since: 2022.9.29 19:22
 * @description:
 */
@Component
public class PromoProducer {
    @Autowired
    PromoItemMapper promoItemMapper;
    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;

    private TransactionMQProducer producer;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("create_promo_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //TODO 秒杀服务的扣除订单
                Map<String, Long> dataMap = (Map<String, Long>) arg;
                Long productId = dataMap.get("productId");
                Long psId = dataMap.get("psId");
                //TODO 加锁的问题
                //锁的在redis中key保证唯一，因为在不同的访问实例中使用的是不同的对象，普通synchronized ()没有作用
                String lockKey ="promo_lock:"+productId+":"+psId;
                RLock promoLock = redissonClient.getLock(lockKey);
                promoLock.lock();
                Integer effctiveRows = 0;
                try {
                    effctiveRows = promoItemMapper.decreaseStock(productId, psId);
                }finally {
                    promoLock.unlock();
                }

                //TODO 设置
                if (effctiveRows < 1) {
                    //TODO 设置库存售罄 减小访问数据库的压力
                    String promoItemSeckillStatus ="promo_seck:"+psId+":"+productId;
                    cacheManager.setCache(promoItemSeckillStatus,"none",1);

                    //TODO 没有扣除成功
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //TODO 成功就是把数据存到redis中
                String transactionKey ="promo+producer"+msg.getTransactionId();
                cacheManager.setCache(transactionKey, "COMMIT_MESSAGE", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String transactionKey ="promo+producer"+msg.getTransactionId();
                String transactionResult = cacheManager.checkCache(transactionKey);
                if (StringUtils.isEmpty(transactionResult)) {
                    return LocalTransactionState.UNKNOW;
                }
                if ("COMMIT_MESSAGE".equals(transactionResult)) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public boolean sendCerateOrder(CreateSeckillOrderRequest request, Long productId, Long psId) {
        HashMap<String, Long> map = new HashMap<>();
        map.put("productId",productId);
        map.put("psId",psId);
        Message message = new Message();
        message.setTopic("create_promo");

        String dataJson = JSON.toJSONString(request);

        message.setBody(dataJson.getBytes(Charset.forName("utf-8")));

        TransactionSendResult transactionSendResult = null;
        try {
            transactionSendResult = producer.sendMessageInTransaction(message, map);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (transactionSendResult != null &&
                LocalTransactionState.COMMIT_MESSAGE.
                        equals(transactionSendResult.getLocalTransactionState())) {
            return true;
        }

        return false;

    }
}
